package com.xtayfjpk.zookeeper.test.master_workers;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;

import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.CreateWorkerAssignCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.CreateWorkerCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.WorkerStatusUpdateCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.WorkerTaskDataCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.WorkerTaskDeletionCreateCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.WorkerTaskStatusCreateCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.worker.WorkerTasksGetChildrenCallback;
import com.xtayfjpk.zookeeper.test.master_workers.watcher.NewTaskWatcher;

public class Worker implements Watcher {
	private static final Logger LOGGER = Logger.getLogger(Master.class);
	private static final String DEFAULT_HOST_PORT = "192.168.0.202:2181";
	private ZooKeeper zk;
	private String hostPort;
	private String serverId = Integer.toHexString(new Random().nextInt());
	private String status;
	private String name;
	
	private volatile boolean connected = false;
    private volatile boolean expired = false;
    
    private ThreadPoolExecutor executor;
	
	public Worker(String hostPort) {
		this.hostPort = hostPort;
		
		this.executor = new ThreadPoolExecutor(1, 1, 
                1000L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(200));
	}
	
	public void startZK() throws IOException {
		zk = new ZooKeeper(hostPort, 60000, this);
	}
	
	
	public void register() {
		zk.create(
				"/workers/worker-" + serverId, 
				"Idle".getBytes(),
				Ids.OPEN_ACL_UNSAFE, 
				CreateMode.EPHEMERAL,
    			new CreateWorkerCallback(this), 
    			null);
	}
	
	public void monitorTasks() {
		zk.getChildren(
				"/assign/worker-" + serverId, 
				new NewTaskWatcher(this), 
				new WorkerTasksGetChildrenCallback(this), 
				null);
	}
	
	public void setStatus(String status) {
		this.status = status;
		updateStatus(status);
	}
	
	synchronized public void updateStatus(String status) {
		//TODO ??????????
		if(status==this.status) {
			zk.setData(
					"/workers/" + name, 
					status.getBytes(), 
					-1,
                    new WorkerStatusUpdateCallback(this), 
                    status);
		}
	}
	
	@Override
	public void process(WatchedEvent event) {
		LOGGER.info(event.toString() + ", " + hostPort);
		
		if(event.getType()==EventType.None) {
			switch (event.getState()) {
				case SyncConnected:
					connected = true;
					break;
				case Disconnected:
					connected = false;
					break;
				case Expired:
					expired = true;
					connected = false;
					LOGGER.error("Session expired");
					break;
				default:
					break;
			}
		}

	}
	
	private void bootstrap() {
		//当然worker启动后，就需要去/assign节点下创建标识自己的子节点
		createAssignNode();		
	}
	
	public void createAssignNode() {
		zk.create(
				"/assign/worker-" + serverId, 
				new byte[0], 
				Ids.OPEN_ACL_UNSAFE, 
				//不能使用EPHEMERAL类型，EPHEMERAL类型不能有子节点
				//并且当会话断开后该节点不能自动消息，因为其下有已经配置的任务，需要重新分配
				CreateMode.PERSISTENT, 
				new CreateWorkerAssignCallback(this), 
				null);
	}
	
	public void getNewTaskData(String task) {
		LOGGER.info("New task: " + task);
		zk.getData(
				"/assign/worker-" + serverId, 
				false, 
				new WorkerTaskDataCallback(this), 
				task);
	}
	
	public void createDoneStatus(String path) {
		zk.create(
				path, 
				"done".getBytes(), 
				Ids.OPEN_ACL_UNSAFE, 
                CreateMode.PERSISTENT, 
                new WorkerTaskStatusCreateCallback(this), 
                null);
	}
	
	public void deleteDoneTask(String path) {
		zk.delete(
				path, 
                -1, 
                new WorkerTaskDeletionCreateCallback(this), 
                null);
	}
	
	public boolean isConnected() {
		return connected;
	}
	
	public boolean isExpired() {
		return expired;
	}
	public String getServerId() {
		return serverId;
	}
	
	public ThreadPoolExecutor getExecutor() {
		return executor;
	}
	
	public ZooKeeper getZk() {
		return zk;
	}
	public static void main(String[] args) throws Exception {
		Worker worker = new Worker(DEFAULT_HOST_PORT);
		worker.startZK();
		
		while(!worker.isConnected()) {
			Thread.sleep(100);
		}
		
		worker.bootstrap();
		
		worker.register();
		
		worker.monitorTasks();
		
		while(!worker.isExpired()) {
			Thread.sleep(100000);
		}
	}



}
